<?php
// +----------------------------------------------------------------------
// | ThinkCMF [ WE CAN DO IT MORE SIMPLE ]
// +----------------------------------------------------------------------
// | Copyright (c) 2013-2014 http://www.thinkcmf.com All rights reserved.
// +----------------------------------------------------------------------
// | Author: Forpastime <forpastime@qq.com>
// +----------------------------------------------------------------------
namespace plugins\rabbitmq\controller; 

use cmf\controller\PluginBaseController;
use think\Db;
use think\db\Query;
use think\facade\Request;

class IndexController extends PluginBaseController
{
    protected $config;
    private $conn;//mq链接
    private $channel;
    private $exchange;
    private $queue;

    private $routeKey = 'abc';

    public function initialize()
    {
        $this->config = $this->getPlugin()->getConfig();
        
        //创建链接
        $conn_args = array( 
            'host' => $this->config['host'],  
            'port' => $this->config['port'],  
            'login' => $this->config['login'],  
            'password' => $this->config['password'], 
            'vhost'=>$this->config['vhost'] 
        ); 
        $conn = new \AMQPConnection($conn_args);
       
        if (!$conn->connect()) {   
            die("Cannot connect to the broker!\n");
        }  
        $this->conn = $conn;

        $this->channel();
    }

    /**
     * 创建连接和channel
     */
    public function channel ()
    {
        $this->channel = new \AMQPChannel($this->conn);  
    }

    /**
     * 创建exchange交换机  
     *  @param exchangeName
     *  @param type queueName 队列名
     */
    public function setExchange ($exchangeName)
    {
        
        $ex = new \AMQPExchange($this->channel);

        $ex->setName($exchangeName); 
        $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
        $ex->setFlags(AMQP_DURABLE); //持久化
        $ex->declareExchange();

        $this->exchange = $ex;  
        
        return $ex;

    }

    /**
     * 创建队列  
     */
    public function setQueue ($queueName,$exchangeName,$routeKey)
    {
        //  创建队列
        $queue = new \AMQPQueue($this->channel); 
        $queue->setName($queueName);   
        $queue->setFlags(AMQP_DURABLE); //持久化 
        $queue->declareQueue();

        // 绑定队列和交换机
        $queue->bind($exchangeName,$routeKey);

        $this->queue = $queue;  

        return $queue;
    }

    /**
    　　* 消费回调函数
    　　* 处理消息
    　　* *需要注意的地方是：queue对象有两个方法可用于取消息： consume 和 get 。
    　　*前者是阻塞的，无消息时会被挂起，适合循环中使用；后者则是非阻塞的，
    　　*取消息时有则取，无则返回false。
    */
    public function processMessage($envelope, $queue) {
        $msg = $envelope->getBody();
        echo $msg."\n"; //处理消息
        $queue->ack($envelope->getDeliveryTag());//手动应答
    }

    /**
     * 消费者订阅
     * param $function 消费回调函数
     */
    public function consumer()
    {
        $param = $this->request->param();
        if(empty($param) || empty($param['queue_name']) || empty($param['exchange_name'])  || empty($param['routing_key'])){
            $this->error('参数异常');
        }
        $queueName = $param['queue_name'];
        $exchangeName = $param['exchange_name'];
        $routingKey = $param['routing_key'];

        $queue = $this->setQueue($queueName,$exchangeName,$routingKey);
        

        //方法一
        $start = time();
        while(true)
        {
            $message = $queue->get();
            if(!empty($message))
            {
                $r = mt_rand(1,20);
                Db::name('tenant_user_category')->where('id',1)->update(['status'=>$r]);

                echo $message->getBody();
                $queue->ack($message->getDeliveryTag());    //应答，代表该消息已经消费
                $end = time();
                echo '<br>' . ($end - $start);
                // sleep(3);
                // exit();
            }
            else
            {
                // $this->conn->disconnect();
                // exit();
                //echo 'message not found' . PHP_EOL;
            }
        }

        //方法二  
        // function callback(AMQPEnvelope $message) {
        //     global $queue;
        //     if ($message) {
        //         $body = $message->getBody();
        //         echo '接收时间：'.date("Y-m-d H:i:s", time()). PHP_EOL;
        //         echo '接收内容：'.$body . PHP_EOL;
        //         //为了防止接收端在处理消息时down掉，只有在消息处理完成后才发送ack消息
        //         $queue->ack($message->getDeliveryTag());
        //     } else {
        //         echo 'no message' . PHP_EOL;
        //     }
        // }
        // $queue->consume(function callback(AMQPEnvelope $message) {
        //     if ($message) {
        //         $body = $message->getBody();
        //         echo '接收时间：'.date("Y-m-d H:i:s", time()). PHP_EOL;
        //         echo '接收内容：'.$body . PHP_EOL;
        //         //为了防止接收端在处理消息时down掉，只有在消息处理完成后才发送ack消息
        //         $queue->ack($message->getDeliveryTag());
        //     } else {
        //         echo 'no message' . PHP_EOL;
        //     }
        // });

        // $queue->consume('callback');  //第一种消费方式,但是会阻塞,程序一直会卡在此处
        
        $this->conn->disconnect();
        
    }

    /**
     * 发布者
     */
    public function producer()
    {
        $param = $this->request->param();

        if(empty($param) || empty($param['exchange_name']) || empty($param['routing_key'])){
            $this->error('参数异常');
        }

        $msg = $param['msg'];
        $exchangeName = $param['exchange_name'];
        $routingKey = $param['routing_key'];

        $ex = $this->setExchange($exchangeName);
        $res = $ex->publish(json_encode($msg),$routingKey, AMQP_NOPARAM, array('delivery_mode' => 2));
        echo "Send Message:".$res;
    }
}